package org.wj.config.avg.learn.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;

import java.util.List;
import java.util.Map;
import java.util.Random;

@Slf4j
public class AuditPartitioner implements Partitioner {

    private Random random;

    @Override
    public int partition(String s, Object o, byte[] bytes, Object o1, byte[] bytes1, Cluster cluster) {
        String key = String.valueOf(o);
        List<PartitionInfo> partitionInfos = cluster.availablePartitionsForTopic(s);
        int size = partitionInfos.size();
        log.debug("分区总数{}:", size);
        int auditPartition = size - 1;
        log.debug("audit分区{}", auditPartition);
        log.info("{}", cluster);
        return key == null || key.isEmpty() || !key.startsWith("audit") ? auditPartition==0 ? auditPartition :random.nextInt(auditPartition) : auditPartition;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {
        random = new Random();
    }
}
